package cloud.juxue.config;

import cloud.juxue.protodto.BinLogInfo;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.Optional;

/**
 * @Description TODO
 * @Date 11:43 2022/1/7
 * @Version 1.0
 * @AUTHOR by yhsh
 */
public class LoggingErrorHandler implements ErrorHandler {

    private static Logger logger = LoggerFactory.getLogger(LoggingErrorHandler.class);

    @Override
    public void handle(Exception e, ConsumerRecord<?, ?> consumerRecord) {

        logger.error("消费信息抛出异常信息--{}",e.getMessage());

       Optional optional =  Optional.ofNullable(consumerRecord.value());
       if (!optional.isPresent()){
            logger.error("监听到的数据为空");
       }
       try {
           if(consumerRecord.value() instanceof byte[]){
               byte[] str = (byte[])consumerRecord.value();
               BinLogInfo.Binlog bin = BinLogInfo.Binlog.parseFrom(str);
               System.out.println(bin.toString());
           }
           logger.error("异常监听数据----------------");
       }catch (Exception ex){
            ex.printStackTrace();
       }


    }
}
